package com.goomba.demo.consumer;


import com.goomba.demo.contants.Commons;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;


/**
 * 1. 默认消费者，Push模式是指我们的MQ主动推送数据到Consumer
 * 2. 我们的生产者只是消息的提供方，他只确定的传递消息到MQ中。具体如何怎么消费由具体的Consumer来确定
 * 由此我们可以指定两种消费模式： 广播和集群。
 * 广播就是每个Consumer(同一个Group下)都会完整的接收到所有的消息。
 * 集群模式下每个Consumer(同一个Group下)会抢占的消费消息。
 * 默认是使用CLUSTERING(集群)模式
 */
public class PushModelConsumer {

    public static void main(String[] args) throws Exception {

        // 1. 创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("con_group1");

        // 2. 指定NameServer
        consumer.setNamesrvAddr(Commons.nameSrvAddr);

        // 3. 消费者订阅具体的topic，并且指定获取的tags
        consumer.subscribe("oneWay", "tag1");

//        consumer.setMessageModel(MessageModel.BROADCASTING);

//        consumer.setMessageModel(MessageModel.CLUSTERING);

        // 4. 注册监听器
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {

            for (int i = 0; i < list.size(); i++) {
                MessageExt message = list.get(i);
                String body = new String(message.getBody());
                System.out.println("返回的数据是" + body);
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 代表消费成功
        });

        consumer.start();

    }

}
